package com.galeno.jobs;

import com.galeno.pojo.LogBean;
import com.galeno.pojo.MatchResult;
import com.galeno.udfs.JSON2LogBean;
import com.galeno.udfs.RuleMatchFunction;
import com.galeno.utils.FlinkUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import java.time.Duration;

/**
 * @author galeno
 * @Title:
 * @Description:用户动态规则的营销系统
 * @date 日期2021/11/14时间14:35
 */
@SuppressWarnings("all")
public class MarketRulesMatch {
    public static void main(String[] args) throws Exception {
        DataStream<String> kafkaStream = FlinkUtils.createKafkaStream(args[0], SimpleStringSchema.class);
        //数据清洗转换

        SingleOutputStreamOperator<LogBean> beanStream = kafkaStream.process(new JSON2LogBean());
        //匹配规则
        SingleOutputStreamOperator<LogBean> logBeanWithWatermark = beanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<LogBean>forBoundedOutOfOrderness(Duration.ofSeconds(0)
        ).withTimestampAssigner(new SerializableTimestampAssigner<LogBean>() {
            @Override
            public long extractTimestamp(LogBean element, long recordTimestamp) {
                return element.getTimeStamp();
            }
        }));


        KeyedStream<LogBean, String> keyedStream = logBeanWithWatermark.keyBy(LogBean::getDeviceId);





        RuleMatchFunction RuleMatchFunction = new RuleMatchFunction();


        SingleOutputStreamOperator<MatchResult> res = keyedStream.process(RuleMatchFunction);
        res.print();
        FlinkUtils.env.execute();
    }
}
